package netty.server;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import kafka.client.KafkaClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.ExecutionException;

public class Server {
    public static final Logger LOG = LoggerFactory.getLogger(Server.class);
    /**
     * server listen port
     */
    private static int portNumber = 7878;
    private static String topicName = "";
    private static String kafkaServerUrl = "";
    public static KafkaClient kafkaClient;
    public static String kafkaMessageKey = "hui-neng";

    public static void main(String[] args) throws InterruptedException {
        if (args.length < 3) {
            System.err.println("Enter socket port, kafka server URL, kafka topic name.");
            System.exit(1);
        }
        portNumber = Integer.valueOf(args[0]);
        topicName = args[2];
        kafkaServerUrl = args[1];
        try {
            kafkaClient = new KafkaClient(topicName, kafkaServerUrl);
        } catch (ExecutionException e) {
            LOG.error("kafka producer can't send message to server: "+kafkaServerUrl);
            e.printStackTrace();
            System.exit(1);
        }

        EventLoopGroup bossGroup = new NioEventLoopGroup();
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        try {
            ServerBootstrap b = new ServerBootstrap();
            b.group(bossGroup, workerGroup);
            b.channel(NioServerSocketChannel.class);
            //b.childHandler(new ServerInitializer());// for byte[]
            b.childHandler(new StringInitializer());//for string
            // bind server port
            ChannelFuture f = b.bind(portNumber).sync();
            // close server listener
            f.channel().closeFuture().sync();
            /* b.bind(portNumber).sync().channel().closeFuture().sync(); */
        } finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }
}